/**
 * Copyright (C), 2015-2018, XXX有限公司
 * FileName: StormKafkaProducerTopo
 * Author:   An-Il
 * Date:     2018/11/18 16:40
 * Description: storm作为生产者
 * History:
 * <author>          <time>          <version>          <desc>
 * 作者姓名           修改时间           版本号              描述
 */
package com.blog.storm.example.integration.kafka.producer;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.kafka.bolt.KafkaBolt;
import org.apache.storm.kafka.bolt.mapper.FieldNameBasedTupleToKafkaMapper;
import org.apache.storm.kafka.bolt.selector.DefaultTopicSelector;
import org.apache.storm.topology.TopologyBuilder;

import java.util.Properties;

/**
 * 〈一句话功能简述〉<br>
 * 〈storm作为生产者,发送消息到kafka〉
 *
 * @author An-Il
 * @create 2018/11/18
 * @since 1.0.0
 */
public class StormKafkaProducerTopo {
    public static void main(String[] args) {

        TopologyBuilder builder = new TopologyBuilder();

        ConsumerSpout consumerSpout = new ConsumerSpout();
        builder.setSpout("consumerSpout", consumerSpout);

        //设置生产者相关参数，详情kafka官网
        Properties props = new Properties();
        props.put("bootstrap.servers", "192.168.31.140:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("linger.ms", 1);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaBolt kafkaBolt = new KafkaBolt()
                .withProducerProperties(props)
                .withTopicSelector(new DefaultTopicSelector("storm_kafka_test"))
                .withTupleToKafkaMapper(new FieldNameBasedTupleToKafkaMapper());
        builder.setBolt("kafkaBolt", kafkaBolt).shuffleGrouping("consumerSpout");

        //本地测试
        LocalCluster localCluster = new LocalCluster();
        localCluster.submitTopology("kafkaboltTest", new Config(), builder.createTopology());

    }
}